package dev.devlok.springcloudstreamzeromqbinder.producers;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.zeromq.SocketType;
import org.zeromq.ZContext;
import org.zeromq.ZMQ;

import static java.util.concurrent.TimeUnit.*;

public class ZeroMqMessageProducer extends MessageProducerSupport {

    private final ConsumerDestination destination;
    private String previousPayload;
    private ZContext zContext;
    private ZMQ.Socket socket;
    private final String socketAddress = "tcp://localhost:6001";

    public ZeroMqMessageProducer(ConsumerDestination destination) {
        this.destination = destination;
    }

    @Override
    public void doStart() {
        receive();
    }

    private void receive() {
        String topic = destination.getName();
        this.zContext = new ZContext();
        this.socket = zContext.createSocket(SocketType.SUB);
        socket.connect(socketAddress);
        socket.subscribe(topic.getBytes(ZMQ.CHARSET));

        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);

        executorService.scheduleWithFixedDelay(() -> {
            String payload = getPayload();

            if (payload != null) {
                Message<String> receivedMessage = MessageBuilder.withPayload(payload).build();
                sendMessage(receivedMessage);
            }

        }, 50, 50, MILLISECONDS);
    }

    private String getPayload() {
        String topic = socket.recvStr();
        String data = socket.recvStr();
        logger.info(topic);
        logger.info(data);
        return data;
    }

    @Override
    protected void doStop() {
        socket.disconnect(socketAddress);
        zContext.destroy();

        super.doStop();
    }
}
